src/parseable/mod.rs (1)

480-489: Consider a builder pattern for stream creation parameters.

The #[allow(clippy::too_many_arguments)] annotation is a reasonable workaround, but with 7+ parameters, this function is becoming unwieldy. For a future improvement, consider introducing a StreamCreationParams struct or builder pattern to group these related parameters.

This is a minor suggestion and doesn't block the current PR.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/parseable/mod.rs` around lines 480 - 489, The function
create_stream_if_not_exists has many parameters; refactor by introducing a
StreamCreationParams struct (or a builder StreamCreationParamsBuilder) to group
stream_name, stream_type, custom_partition, log_source, telemetry_type,
dataset_tags, and dataset_labels; update create_stream_if_not_exists signature
to accept a single StreamCreationParams (or builder output) and adapt internal
usage accordingly, add a conversion constructor or builder methods for easy
construction where callers currently pass those seven+ args, and preserve
existing behavior and validation inside the new struct or builder.

src/parseable/mod.rs (1)

480-527: Consider bundling stream creation parameters into a struct.

The function now accepts 8 parameters (including self), which triggers the clippy::too_many_arguments lint. While the #[allow] attribute suppresses it, this pattern repeats across create_stream_if_not_exists, create_stream, and create_update_stream.

A dedicated struct (e.g., StreamCreationParams or StreamConfig) would:

• Improve readability at call sites
• Make future parameter additions easier
• Provide a natural place for default values

💡 Example struct-based approach

pub struct StreamCreationParams {
    pub stream_type: StreamType,
    pub custom_partition: Option<String>,
    pub log_source: Vec<LogSourceEntry>,
    pub telemetry_type: TelemetryType,
    pub dataset_tags: Vec<DatasetTag>,
    pub dataset_labels: Vec<String>,
}

impl Default for StreamCreationParams {
    fn default() -> Self {
        Self {
            stream_type: StreamType::UserDefined,
            custom_partition: None,
            log_source: vec![],
            telemetry_type: TelemetryType::Logs,
            dataset_tags: vec![],
            dataset_labels: vec![],
        }
    }
}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/parseable/mod.rs` around lines 480 - 527, Create a StreamCreationParams
(or StreamConfig) struct to bundle the multiple stream creation arguments and
refactor create_stream_if_not_exists, create_stream, and create_update_stream to
accept that struct instead of the long parameter list; update the call site in
create_stream_if_not_exists to build a StreamCreationParams (using Default for
sensible defaults) and pass it to create_stream, adjust create_stream signature
to destructure or reference the struct fields (stream_type, custom_partition,
log_source, telemetry_type, dataset_tags, dataset_labels), remove the
#[allow(clippy::too_many_arguments)] on those functions, and add
conversions/constructors where needed so existing call sites can migrate with
minimal changes.

src/handlers/http/datasets.rs (2)

150-156: ⚠️ Potential issue | 🟡 Minor

Normalize label input in PUT body before dedupe.

Line 150-Line 156 deduplicates raw strings but does not trim/filter whitespace-only labels, so empty labels can still be persisted.

Suggested fix

     let final_labels = match body.labels {
         Some(labels) => labels
             .into_iter()
+            .map(|label| label.trim().to_string())
+            .filter(|label| !label.is_empty())
             .collect::<HashSet<_>>()
             .into_iter()
             .collect(),
         None => stream.get_dataset_labels(),
     };

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 150 - 156, Normalize and filter
incoming labels before deduplication: when building final_labels from
body.labels (the match arm handling Some(labels)), map each label to trimmed
string and filter out labels that are empty or whitespace-only (e.g.,
labels.into_iter().map(|s| s.trim().to_string()).filter(|s| !s.is_empty())...),
then collect into a HashSet to dedupe and back into the desired collection; keep
the None branch using stream.get_dataset_labels() unchanged.

---

142-172: ⚠️ Potential issue | 🔴 Critical

Concurrent partial metadata updates can still lose data.

Line 142-Line 172 does a read-modify-write using current in-memory values for absent fields, so two concurrent requests (tags-only and labels-only) can overwrite each other with stale counterparts.

Suggested direction

 pub async fn put_dataset_metadata(...) -> Result<HttpResponse, DatasetsError> {
     ...
     let stream = PARSEABLE
         .get_stream(&dataset_name, &tenant_id)
         .map_err(|_| DatasetsError::DatasetNotFound(dataset_name.clone()))?;
+
+    // Serialize metadata updates for this stream (or use atomic storage merge).
+    // let _guard = stream.metadata_lock().await;

     let final_tags = match body.tags { ... };
     let final_labels = match body.labels { ... };

     storage
         .update_dataset_tags_and_labels_in_stream(...)
         .await
         .map_err(DatasetsError::Storage)?;

     stream.set_dataset_tags(final_tags.clone());
     stream.set_dataset_labels(final_labels.clone());

If a stream-scoped lock is unavailable, prefer a storage-level atomic patch API that updates only provided fields server-side.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 142 - 172, Current code computes
final_tags/final_labels from in-memory stream values and then does a
read-modify-write, which lets concurrent requests overwrite each other; instead
either acquire the stream-scoped lock around the read-modify-write (use
stream.get_dataset_tags/get_dataset_labels and
stream.set_dataset_tags/set_dataset_labels inside the lock) or, preferably,
change the update to an atomic storage-side patch that accepts
Option<HashSet<_>> and only updates provided fields (modify the call to
storage.update_dataset_tags_and_labels_in_stream to pass Option types and
implement server-side merge), removing the reliance on in-memory reads to
prevent lost updates.

src/parseable/streams.rs (1)

982-988: Prefer atomic update for tags+labels to avoid transient mixed state.

Line 982 and Line 986 update related metadata in two separate writes. A single setter that updates both fields under one lock would avoid short-lived inconsistent reads.

♻️ Suggested refactor

+    pub fn set_dataset_metadata(&self, tags: Vec<DatasetTag>, labels: Vec<String>) {
+        let mut metadata = self.metadata.write().expect(LOCK_EXPECT);
+        metadata.dataset_tags = tags;
+        metadata.dataset_labels = labels;
+    }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/parseable/streams.rs` around lines 982 - 988, The two setters
set_dataset_tags and set_dataset_labels perform separate metadata.write() calls
causing transient inconsistent state; add a new atomic setter (e.g.,
set_dataset_tags_and_labels(&self, tags: Vec<DatasetTag>, labels: Vec<String>))
that acquires metadata.write() once and assigns both dataset_tags and
dataset_labels inside the same lock, then update call sites to use the new
method (or have the existing setters delegate to it) so modifications never
occur in two separate writes; reference the metadata RwLock field and the
existing set_dataset_tags/set_dataset_labels method names when making the
change.

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

src/handlers/http/datasets.rs (2)

197-202: ⚠️ Potential issue | 🟡 Minor

Normalize request-body labels before storing them.

This path still stores label bodies verbatim, so blank labels and whitespace variants can persist and skew correlation results.

Suggested fix

         Some(labels) => labels
             .into_iter()
+            .map(|label| label.trim().to_string())
+            .filter(|label| !label.is_empty())
             .collect::<HashSet<_>>()
             .into_iter()
             .collect(),

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 197 - 202, The current
final_labels assignment uses body.labels verbatim and can retain blank or
whitespace-only variants; update the transformation so that when handling
Some(labels) you trim each label (e.g., call trim on the string), discard empty
results (filter out labels where trimmed.is_empty()), and then deduplicate by
collecting into a HashSet before converting back to the Vec used by
final_labels; apply this change to the final_labels computation that consumes
body.labels so stored labels are normalized (trimmed and non-empty).

---

189-216: ⚠️ Potential issue | 🔴 Critical

Partial PUTs still have a lost-update race.

Combining tags and labels into one storage call does not make this atomic because omitted fields are still read from current in-memory state first. A concurrent {"tags": ...} request and {"labels": ...} request can each write back a stale copy of the other field and clobber one update. Either make PUT a full replacement that requires both arrays, or serialize this read/merge/write per stream.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 189 - 216, The current PUT
handler builds final_tags and final_labels by reading in-memory via
stream.get_dataset_tags()/get_dataset_labels() and then calls
storage.update_dataset_tags_and_labels_in_stream(...), which creates a
lost-update race when concurrent partial PUTs update only one field; to fix,
either require full replacement (validate that body.tags and body.labels are
both Some and reject partial PUTs) or serialize read/merge/write per stream
(introduce a per-stream mutex/lock when computing final_tags/final_labels and
calling update_dataset_tags_and_labels_in_stream) so concurrent handlers cannot
interleave; locate and modify the code that constructs final_tags/final_labels
and the call to update_dataset_tags_and_labels_in_stream to implement one of
these two strategies, using identifiers final_tags, final_labels,
stream.get_dataset_tags/get_dataset_labels, and
update_dataset_tags_and_labels_in_stream to find the right spot.

src/handlers/mod.rs (2)

153-161: Same non-deterministic ordering concern for parse_dataset_labels.

Similar to parse_dataset_tags, the label ordering is non-deterministic after HashSet deduplication. This may affect API response consistency.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/mod.rs` around lines 153 - 161, parse_dataset_labels currently
deduplicates via HashSet which yields non-deterministic ordering; change it to
produce a deterministic, stable order by using either a BTreeSet for
deduplication or collect into a Vec then sort before returning. Update the
function parse_dataset_labels (and mirror the same approach used for
parse_dataset_tags if present) to trim and filter as now, then dedupe
deterministically (BTreeSet::from_iter or dedupe Vec + sort) and return a
Vec<String> with a stable sorted order.

---

131-151: Non-deterministic ordering in parse_dataset_tags.

The function collects into a HashSet for deduplication, then converts to Vec. This results in non-deterministic ordering of the returned tags. If consistent ordering matters for API responses or comparison purposes, consider sorting or using IndexSet from the indexmap crate.

// Current: non-deterministic order
.collect::<HashSet<_>>()
.into_iter()
.collect()

// Alternative: deterministic order
.collect::<HashSet<_>>()
.into_iter()
.sorted() // requires Ord impl or sorted_by
.collect()

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/mod.rs` around lines 131 - 151, parse_dataset_tags currently
deduplicates via a HashSet then collects into a Vec which yields
non-deterministic ordering; change it to produce a deterministic order by either
(a) collecting into the HashSet for dedupe and then sorting the iterator
(requiring DatasetTag: Ord or using a comparator) before collecting into Vec, or
(b) replace the HashSet with an IndexSet from the indexmap crate to preserve
insertion order; update the function parse_dataset_tags (and any uses)
accordingly to ensure stable, deterministic ordering of the returned
Vec<DatasetTag>.

src/handlers/http/datasets.rs (2)

197-203: ⚠️ Potential issue | 🟡 Minor

Normalize labels before deduplication and persistence.

Line 197-203 dedupes labels but does not trim/filter empty values, so whitespace-only labels can still be stored and matched inconsistently.

💡 Proposed fix

-    let final_labels = match body.labels {
-        Some(labels) => labels
-            .into_iter()
-            .collect::<HashSet<_>>()
-            .into_iter()
-            .collect(),
+    let final_labels = match body.labels {
+        Some(labels) => labels
+            .into_iter()
+            .map(|label| label.trim().to_string())
+            .filter(|label| !label.is_empty())
+            .collect::<HashSet<_>>()
+            .into_iter()
+            .collect(),
         None => stream.get_dataset_labels(),
     };

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 197 - 203, final_labels is
deduped but not normalized, allowing whitespace-only or untrimmed labels to
persist; update the handling of body.labels (and the branch that falls back to
stream.get_dataset_labels() if needed) to map each label through trim(), filter
out labels that are empty after trimming, then collect into a HashSet to
deduplicate and finally into the expected collection type used later (use the
same symbol final_labels). Ensure both the Some(labels) branch and any code path
that constructs labels from stream.get_dataset_labels() apply the same
trim+filter+dedupe normalization.

---

68-70: ⚠️ Potential issue | 🟠 Major

Defer DatasetNotFound until storage hydration is attempted.

Line 68-70 and Line 185-187 map in-memory get_stream misses straight to 404. In query/standalone flows, this can incorrectly 404 datasets that exist in object storage but are not hydrated yet. Please apply the same storage-hydration existence pattern used in logstream handlers before returning not found.

Based on learnings: stream existence checks should attempt storage-backed hydration in query/standalone mode before returning not found.

Also applies to: 185-187

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 68 - 70, The current code calls
PARSEABLE.get_stream(&dataset_name, &tenant_id) and immediately maps any error
to DatasetsError::DatasetNotFound (e.g., in the block around
PARSEABLE.get_stream and the similar check at lines 185-187), which causes a
premature 404 for datasets that exist in object storage but aren't hydrated;
instead, follow the logstream handler pattern: on get_stream error, attempt
storage-backed hydration (invoke the same hydrate/storage-check routine used by
logstream handlers for query/standalone flows) and only return
DatasetsError::DatasetNotFound if hydration confirms the dataset is absent or
hydration fails definitively; update the code paths around PARSEABLE.get_stream
and the duplicated check to defer mapping to DatasetNotFound until after the
hydration attempt completes.

src/prism/home/mod.rs (1)

63-66: Keep tags and labels in the serialized shape even when empty.

With skip_serializing_if = "Vec::is_empty", datasets that simply have no metadata look the same as responses from servers that do not support these fields at all. Returning [] keeps the home-response schema stable for clients.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/prism/home/mod.rs` around lines 63 - 66, The struct fields `tags:
Vec<DatasetTag>` and `labels: Vec<String>` in src/prism/home/mod.rs should
always be present in serialized output even when empty; remove the
`skip_serializing_if = "Vec::is_empty"` attribute while keeping
`#[serde(default)]` so empty vectors serialize as `[]`. Locate the `tags` and
`labels` field definitions in the struct (the lines with `tags: Vec<DatasetTag>`
and `labels: Vec<String>`) and delete only the `skip_serializing_if` portion of
their serde attributes, leaving `default` intact to still provide an empty Vec
when missing during deserialization.

src/parseable/streams.rs (1)

982-988: Add a single setter for dataset metadata.

The new dataset update flow has to call set_dataset_tags and set_dataset_labels separately, so readers can observe a mixed state between the two writes. A combined setter would keep the in-memory view coherent and avoid the second lock acquisition.

♻️ Suggested refactor

+    pub fn set_dataset_metadata(&self, tags: Vec<DatasetTag>, labels: Vec<String>) {
+        let mut metadata = self.metadata.write().expect(LOCK_EXPECT);
+        metadata.dataset_tags = tags;
+        metadata.dataset_labels = labels;
+    }
+
     pub fn set_dataset_tags(&self, tags: Vec<DatasetTag>) {
         self.metadata.write().expect(LOCK_EXPECT).dataset_tags = tags;
     }
 
     pub fn set_dataset_labels(&self, labels: Vec<String>) {
         self.metadata.write().expect(LOCK_EXPECT).dataset_labels = labels;
     }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/parseable/streams.rs` around lines 982 - 988, Introduce a single atomic
setter that updates both tags and labels under one write lock to avoid
intermediate inconsistent reads: add a new method (e.g. set_dataset_metadata or
set_dataset_tags_and_labels) that takes (tags: Vec<DatasetTag>, labels:
Vec<String>) and performs one self.metadata.write().expect(LOCK_EXPECT)
assignment setting both dataset_tags and dataset_labels in the same critical
section, and update call sites to use this combined setter instead of calling
set_dataset_tags and set_dataset_labels separately; keep the existing individual
setters only if backward compatibility is required but prefer the combined
setter to prevent two separate lock acquisitions and mixed-state visibility.

src/handlers/mod.rs (1)

131-151: Non-deterministic ordering in parsed tags/labels.

The parse_dataset_tags and parse_dataset_labels functions collect into a HashSet for deduplication then convert to Vec. Since HashSet iteration order is non-deterministic, the resulting vector order may vary across calls with the same input. This could cause issues with:

• Consistent API responses
• Testing
• Diff-based change detection

If ordering matters for any downstream use case (e.g., serialization stability), consider sorting the result or using IndexSet from the indexmap crate to preserve insertion order while deduplicating.

♻️ Optional fix to ensure deterministic ordering

 pub fn parse_dataset_tags(header_value: &str) -> Vec<DatasetTag> {
-    header_value
+    let mut tags: Vec<DatasetTag> = header_value
         .split(',')
         .filter_map(|s| {
             let trimmed = s.trim();
             if trimmed.is_empty() {
                 None
             } else {
                 match DatasetTag::try_from(trimmed) {
                     Ok(tag) => Some(tag),
                     Err(err) => {
                         warn!("Invalid dataset tag '{trimmed}': {err}");
                         None
                     }
                 }
             }
         })
         .collect::<HashSet<_>>()
         .into_iter()
-        .collect()
+        .collect();
+    tags.sort_by(|a, b| a.to_string().cmp(&b.to_string()));
+    tags
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/mod.rs` around lines 131 - 151, The functions parse_dataset_tags
(and parse_dataset_labels) deduplicate via HashSet then collect into Vec, which
yields non-deterministic ordering; update these functions to produce
deterministic output by either (a) replacing the HashSet dedupe with an IndexSet
from the indexmap crate to preserve insertion order while removing duplicates,
or (b) after collecting into a Vec, sort the Vec deterministically (e.g.,
lexicographically by tag/label string or by a stable Ord implementation) before
returning; apply the same change to both parse_dataset_tags and
parse_dataset_labels so repeated calls with the same input yield a stable
ordering.

src/handlers/http/datasets.rs (2)

211-217: ⚠️ Potential issue | 🟡 Minor

Normalize labels before deduplication (trim + drop empty).

Current label handling can persist empty/whitespace-only labels and produce noisy correlation results.

Proposed fix

     let final_labels = match body.labels {
         Some(labels) => labels
             .into_iter()
+            .map(|label| label.trim().to_string())
+            .filter(|label| !label.is_empty())
             .collect::<HashSet<_>>()
             .into_iter()
             .collect(),
         None => stream.get_dataset_labels(),
     };

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 211 - 217, The label handling for
final_labels (when reading body.labels or falling back to
stream.get_dataset_labels()) must normalize values before deduplication: trim
whitespace from each label and drop any empty/whitespace-only strings, then
collect into a HashSet to dedupe and back into the final collection. Update the
Some(labels) branch (body.labels) to map each label through trim + to_string,
filter out empties, then dedupe; and apply the same normalization/filtering to
the fallback stream.get_dataset_labels() result so both paths produce cleaned,
non-empty, deduplicated labels.

---

203-218: ⚠️ Potential issue | 🔴 Critical

Partial-update merge is still vulnerable to lost updates under concurrency.

None => existing + single write means concurrent requests (one tags-only, one labels-only) can overwrite each other’s field with stale values. This is still a TOCTOU/lost-write bug.

Suggested direction

Serialize the read-merge-write path per stream (mutex/lock), or move merge semantics into storage as an atomic patch operation so omitted fields are preserved against the latest persisted state.

Also applies to: 220-233

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 203 - 218, The current
read-merge-write for final_tags/final_labels uses body.tags/body.labels with
fallback to stream.get_dataset_tags()/get_dataset_labels(), which allows TOCTOU
lost-updates under concurrent partial updates; to fix, serialize the
read-merge-write for a given stream (e.g., acquire a per-stream mutex/lock
around reading stream.get_dataset_*(), merging with body.tags/body.labels, and
writing) or implement an atomic patch in storage that merges omitted fields
server-side so updates to tags and labels cannot overwrite each other; update
the code paths that compute final_tags and final_labels (and the duplicated
logic at lines ~220-233) to use the chosen serialized/atomic approach.

src/handlers/http/datasets.rs (2)

213-240: ⚠️ Potential issue | 🔴 Critical

Partial PUTs can still clobber each other under concurrency.

Combining tags and labels into one endpoint removed the cross-endpoint race, but the None => stream.get_dataset_*() merge keeps a read-modify-write window here. If one request updates only tags while another updates only labels, both can read the same old state and the later write will restore one field to a stale value. Please serialize metadata updates per stream, or move the merge into a single atomic storage-side update/CAS.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 213 - 240, The current
read-modify-write uses body.tags/body.labels with
stream.get_dataset_tags()/stream.get_dataset_labels() then calls
update_dataset_tags_and_labels_in_stream, which allows concurrent partial-PUTs
to clobber each other; fix by moving the merge/compare-and-swap into the storage
layer (make update_dataset_tags_and_labels_in_stream perform an atomic
read-merge-CAS using tenant_id and dataset_name) or by adding per-stream
serialization around this handler (e.g., a per-stream mutex) so updates to
final_tags/final_labels cannot interleave; modify the storage method or
introduce a per-stream lock and remove the local stream.get_dataset_* merge to
ensure atomicity.

---

221-227: ⚠️ Potential issue | 🟡 Minor

Trim and drop blank labels before persisting.

This path still accepts "", whitespace-only labels, and " foo " as distinct labels, so correlation can diverge from the header-based creation flow.

Suggested fix

     let final_labels = match body.labels {
         Some(labels) => labels
             .into_iter()
+            .map(|label| label.trim().to_string())
+            .filter(|label| !label.is_empty())
             .collect::<HashSet<_>>()
             .into_iter()
             .collect(),
         None => stream.get_dataset_labels(),
     };

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/handlers/http/datasets.rs` around lines 221 - 227, Normalize incoming
labels by trimming whitespace and dropping empty strings before collecting into
the final set: in the Some(labels) branch where final_labels is built, replace
the current collect pipeline with one that maps each label through
trim().to_string(), filters out empty strings (after trim), then collects into a
HashSet to deduplicate and back into the desired collection. Also ensure the
same normalization is applied if you use stream.get_dataset_labels() or any
other source so header-based and body-based flows match.